package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.OrderWide;
import com.atguigu.realtime.bean.PaymentWide;
import com.atguigu.realtime.bean.ProductStats;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.function.DimAsyncFunctioin;
import com.atguigu.realtime.util.CommonUtil;
import com.atguigu.realtime.util.DimUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

import static com.atguigu.realtime.common.Constant.*;

/**
 * @author chenlongStart
 * @create 2021-07-03 9:47
 * @desc
 */
public class DwsProductStatsApp extends BaseAppV2 {

    public static void main(String[] args) {
        new DwsProductStatsApp().init(
                4002, 1,
                "DwsProductStatsApp",
                TOPIC_DWD_PAGE_LOG,
                "DwsProductStatsApp",
                TOPIC_DWD_DISPLAY_LOG,
                TOPIC_DWD_FAVOR_INFO,
                TOPIC_DWD_CART_INFO,
                TOPIC_DWM_ORDER_WIDE,
                TOPIC_DWM_PAYMENT_WIDE,
                TOPIC_DWD_ORDER_REFUND_INFO,
                TOPIC_DWD_COMMENT_INFO
        );
    }

    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> dsMap) {
        // 1、8个流union到一起
        DataStream<ProductStats> productStatsStream = parseAndUnion(dsMap);

        // 2、分组开窗聚合
        SingleOutputStreamOperator<ProductStats> reducedStatsStream = reduceByWindowAndSkuId(productStatsStream);

        // 3、补充维度信息
        SingleOutputStreamOperator<ProductStats> reducedStreamWithDim = joinDim(reducedStatsStream);

        // 4、写入clickhouse
        sendToClickhouse(reducedStreamWithDim);

        // 5. 由于产品行为关键词的需要, 把这个数据写入到kafka
        sendToKafka(reducedStreamWithDim);
    }

    private void sendToKafka(SingleOutputStreamOperator<ProductStats> reducedStreamWithDim) {
        reducedStreamWithDim
                .map(JSON::toJSONString)
                .addSink(FlinkSinkUtil.getKafkaSink(Constant.TOPIC_DWS_PRODUCT_STATS));
    }

    private void sendToClickhouse(SingleOutputStreamOperator<ProductStats> stream) {
        stream.addSink(FlinkSinkUtil.getClickhouseSink("gmall2021", "product_stats_2021", ProductStats.class));
    }

    private SingleOutputStreamOperator<ProductStats> joinDim(SingleOutputStreamOperator<ProductStats> aggregatedStream) {
        return AsyncDataStream.unorderedWait(
                aggregatedStream,
                new DimAsyncFunctioin<ProductStats>() {
                    @Override
                    public void addDim(Connection conn,
                                       Jedis jedis,
                                       ProductStats input,
                                       ResultFuture<ProductStats> resultFuture) throws Exception {
                        // 1. 补齐sku信息, 和 spu_id, tm_id c3_id
                        JSONObject skuInfo = DimUtil.readDim(conn, jedis, DIM_SKU_INFO, input.getSku_id());
                        input.setSku_name(skuInfo.getString("SKU_NAME"));
                        input.setSku_price(skuInfo.getBigDecimal("PRICE"));

                        input.setSpu_id(skuInfo.getLong("SPU_ID"));
                        input.setTm_id(skuInfo.getLong("TM_ID"));
                        input.setCategory3_id(skuInfo.getLong("CATEGORY3_ID"));

                        // 2. 补齐spu信息
                        JSONObject spuInfo = DimUtil.readDim(conn, jedis, DIM_SPU_INFO, input.getSpu_id());
                        input.setSpu_name(spuInfo.getString("SPU_NAME"));
                        // 3. 补齐tm信息
                        JSONObject tmInfo = DimUtil.readDim(conn, jedis, DIM_BASE_TRADEMARK, input.getTm_id());
                        input.setTm_name(tmInfo.getString("TM_NAME"));
                        // 3. 补齐c3信息
                        JSONObject c3Info = DimUtil.readDim(conn, jedis, DIM_BASE_CATEGORY3, input.getCategory3_id());
                        input.setCategory3_name(c3Info.getString("NAME"));

                        resultFuture.complete(Collections.singletonList(input));
                    }
                },
                300,
                TimeUnit.SECONDS);
    }

    private SingleOutputStreamOperator<ProductStats> reduceByWindowAndSkuId(DataStream<ProductStats> productStatsStream) {
        SingleOutputStreamOperator<ProductStats> result = productStatsStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                                .withTimestampAssigner((ps, ts) -> ps.getTs())
                )
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<ProductStats>("late") {
                })
                .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats v1,
                                                       ProductStats v2) throws Exception {
                                v1.setClick_ct(v1.getClick_ct() + v2.getClick_ct());
                                v1.setDisplay_ct(v1.getDisplay_ct() + v2.getDisplay_ct());
                                v1.setFavor_ct(v1.getFavor_ct() + v2.getFavor_ct());
                                v1.setCart_ct(v1.getCart_ct() + v2.getCart_ct());
                                v1.setComment_ct(v1.getComment_ct() + v2.getComment_ct());
                                v1.setGood_comment_ct(v1.getGood_comment_ct() + v2.getGood_comment_ct());

                                v1.setOrder_amount(v1.getOrder_amount().add(v2.getOrder_amount()));
                                v1.setOrder_sku_num(v1.getOrder_sku_num() + v2.getOrder_sku_num());
                                v1.setPayment_amount(v1.getPayment_amount().add(v2.getPayment_amount()));
                                v1.setRefund_amount(v1.getRefund_amount().add(v2.getRefund_amount()));


                                v1.getOrderIdSet().addAll(v2.getOrderIdSet());
                                v1.getRefundOrderIdSet().addAll(v2.getRefundOrderIdSet());
                                v1.getPaidOrderIdSet().addAll(v2.getPaidOrderIdSet());
                                return v1;
                            }
                        }, new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {

                            private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<ProductStats> elements,
                                                Collector<ProductStats> out) throws Exception {
                                ProductStats ps = elements.iterator().next();
                                TimeWindow w = ctx.window();
                                ps.setStt(sdf.format(w.getStart()));
                                ps.setEdt(sdf.format(w.getEnd()));

                                // 和订单相关的3个count进行计算
                                ps.setOrder_ct((long)ps.getOrderIdSet().size());
                                ps.setRefund_order_ct((long)ps.getRefundOrderIdSet().size());
                                ps.setPaid_order_ct((long)ps.getPaidOrderIdSet().size());

                                out.collect(ps);
                            }
                        }
                );
        return result;
    }

    private DataStream<ProductStats> parseAndUnion(HashMap<String, DataStreamSource<String>> dsMap) {
        // 1、获取点击量
        SingleOutputStreamOperator<ProductStats> clickStatsStream = dsMap
                .get(TOPIC_DWD_PAGE_LOG)
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String jsonStr,
                                               Context ctx,
                                               Collector<ProductStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(jsonStr);
                        JSONObject page = obj.getJSONObject("page");
                        String pageId = page.getString("page_id");
                        if ("good_detail".equals(pageId)) {
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(page.getLong("item"));
                            ps.setTs(obj.getLong("ts"));
                            ps.setClick_ct(1L);
                            out.collect(ps);
                        }
                    }
                });

        //2、获取曝光量
        SingleOutputStreamOperator<ProductStats> displayStatsStream = dsMap
                .get(TOPIC_DWD_DISPLAY_LOG)
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String jsonStr,
                                               Context ctx,
                                               Collector<ProductStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(jsonStr);
                        if ("sku_id".equals(obj.getString("item_type"))) {
                            ProductStats ps = new ProductStats();
                            ps.setSku_id(obj.getLong("item"));
                            ps.setTs(obj.getLong("ts"));
                            ps.setDisplay_ct(1L);
                            out.collect(ps);
                        }
                    }
                });

        //3、获取收藏量
        SingleOutputStreamOperator<ProductStats> favorStatsStream = dsMap
                .get(TOPIC_DWD_FAVOR_INFO)
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(CommonUtil.toTs(obj.getString("create_time")));
                    ps.setFavor_ct(1L);
                    return ps;
                });

        //4、获取加入购物车量
        SingleOutputStreamOperator<ProductStats> cartStatsStream = dsMap
                .get(TOPIC_DWD_CART_INFO)
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(CommonUtil.toTs(obj.getString("create_time")));
                    ps.setCart_ct(1L);
                    return ps;
                });

        //5、获取下单量
        SingleOutputStreamOperator<ProductStats> orderStatsStream = dsMap
                .get(TOPIC_DWM_ORDER_WIDE)
                .map(jsonStr -> {
                    OrderWide ow = JSON.parseObject(jsonStr, OrderWide.class);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(ow.getSku_id());
                    ps.setTs(CommonUtil.toTs(ow.getCreate_time()));
                    ps.setOrder_amount(ow.getSplit_total_amount());//这个商品在这个订单里占的金额
                    ps.setOrder_sku_num(ow.getSku_num());
                    ps.getOrderIdSet().add(ow.getOrder_id());//这个用来统计这个商品别多个订单下过
                    return ps;
                });

        //6、获取支付量
        SingleOutputStreamOperator<ProductStats> paymentStatsStream = dsMap
                .get(TOPIC_DWM_PAYMENT_WIDE)
                .map(jsonStr -> {
                    PaymentWide pw = JSON.parseObject(jsonStr, PaymentWide.class);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(pw.getSku_id());
                    ps.setTs(CommonUtil.toTs(pw.getPayment_create_time()));
                    ps.setPayment_amount(pw.getSplit_total_amount());
                    ps.getPaidOrderIdSet().add(pw.getOrder_id());
                    return ps;
                });

        //7、获取退款量
        SingleOutputStreamOperator<ProductStats> refundStatsStream = dsMap
                .get(TOPIC_DWD_ORDER_REFUND_INFO)
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(CommonUtil.toTs(obj.getString("create_time")));
                    ps.setRefund_amount(obj.getBigDecimal("refund_amount"));
                    ps.getRefundOrderIdSet().add(obj.getLong("order_id"));
                    return ps;
                });

        //8、评论
        SingleOutputStreamOperator<ProductStats> commentStatsStream = dsMap
                .get(TOPIC_DWD_COMMENT_INFO)
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    ProductStats ps = new ProductStats();
                    ps.setSku_id(obj.getLong("sku_id"));
                    ps.setTs(CommonUtil.toTs(obj.getString("create_time")));
                    ps.setComment_ct(1L);
                    if (FIVE_STAR_GOOD_COMMENT.equals(obj.getString("appraise"))
                            || FOUR_STAR_GOOD_COMMENT.equals(obj.getString("appraise"))) {
                        ps.setGood_comment_ct(1L);
                    }
                    return ps;
                });

        //9、Union
        return clickStatsStream
                .union(
                        displayStatsStream,
                        favorStatsStream,
                        cartStatsStream,
                        orderStatsStream,
                        paymentStatsStream,
                        refundStatsStream,
                        commentStatsStream
                );
    }

}
